package com.freez.spark.core.demo;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.*;
import scala.Tuple2;

import java.util.Arrays;
import java.util.Iterator;

/**
 * FREEDOM  2021 人生苦短，不妨一试
 * FileName: SumDemo.java
 * Author: zcs
 * Date: 2021年-12月-07日 周二 18:35
 * Description: 词频统计
 */
public class WordCount {
    public static void main(String[] args) {
        SparkConf conf = new SparkConf();
        conf.setAppName("spark-firstDemo").setMaster("local");
        JavaSparkContext jsc = new JavaSparkContext(conf);
        JavaRDD<String> jRDD = jsc.textFile("wtf");
        // 切片分离，扁平化
        JavaRDD<String> wordRDD = jRDD.flatMap(new FlatMapFunction<String, String>() {
            public Iterator<String> call(String line) throws Exception {
                return Arrays.asList(line.split(" ")).iterator();
            }
        });
        //TODO: 组装kv,Tuple2.apply(word, 1),第一个表示单词，第二个表示出现次数
        JavaPairRDD<String, Integer> kvRDD = wordRDD.mapToPair(new PairFunction<String, String, Integer>() {
            //Tuple2:二元组
            public Tuple2<String, Integer> call(String word) throws Exception {
                Tuple2<String, Integer> apply = Tuple2.apply(word, 1);
                return apply;
            }
        });
        //TODO:分组聚合，分组累加计数
        JavaPairRDD<String, Integer> countRDD = kvRDD.reduceByKey(new Function2<Integer, Integer, Integer>() {
            public Integer call(Integer v1, Integer v2) throws Exception {
                return v1 + v2;
            }
        });
        /**
         * kv互换
         */
//        JavaPairRDD<Integer, String> swapRDD = kvRDD.mapToPair(new PairFunction<Tuple2<String, Integer>, Integer, String>() {
//            public Tuple2<Integer, String> call(Tuple2<String, Integer> tp) throws Exception {
//                return tp.swap();
//            }
//        });
//
//
//        JavaRDD<Object> map = swapRDD.map(new Function<Tuple2<Integer, String>, Object>() {
//            public Object call(Tuple2<Integer, String> v1) throws Exception {
//                return null;
//            }
//        });
//        countRDD.sortByKey();
        countRDD.foreach(new VoidFunction<Tuple2<String, Integer>>() {
            public void call(Tuple2<String, Integer> s) throws Exception {
                System.out.println(s);
            }
        });


        JavaRDD<Integer> demoRDD = jsc.parallelize(Arrays.asList(1, 4, 6));
        JavaPairRDD<String, Integer> mapRDD = demoRDD.mapToPair(new PairFunction<Integer, String, Integer>() {
            @Override
            public Tuple2<String, Integer> call(Integer integer) throws Exception {
                Tuple2<String, Integer> tuple2 = new Tuple2<String, Integer>("Key" + integer, integer);
                System.out.println("mapToPair--------" + tuple2);
                return tuple2;
            }
        });

        JavaPairRDD<String, Integer> filterRDD = mapRDD.filter(new Function<Tuple2<String, Integer>, Boolean>() {
            @Override
            public Boolean call(Tuple2<String, Integer> v1) throws Exception {
                System.out.println("filter--------" + v1);
                return v1._2 != 6;
            }
        });
        filterRDD.foreach(new VoidFunction<Tuple2<String, Integer>>() {
            @Override
            public void call(Tuple2<String, Integer> s) throws Exception {
                System.out.println("result--------" + s);
            }
        });
        jsc.close();
    }
}
